package com.atguigu.transform;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.checkerframework.checker.units.qual.C;


/**
 * @Author Mr.Zheng
 * @Date 2023/6/14 18:56
 *
 * 分流：
 *      1. 简单实现  通过filter算子来实现
 *      2. 使用侧输出流
 *
 */
public class Flink08_SplitStream {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        DataStreamSource<Event> ds = env.addSource(new ClickSource());


        // 将不同用户的数据分到不同流中
        // 简单实现：filter

        /*
        SingleOutputStreamOperator<Event> ZhangsDs = ds.filter(event -> "Zhangs".equals(event.getUser()));
        SingleOutputStreamOperator<Event> LisiDs = ds.filter(event -> "Lisi".equals(event.getUser()));
        SingleOutputStreamOperator<Event> TomDs = ds.filter(event -> "Tom".equals(event.getUser()));
        SingleOutputStreamOperator<Event> JerryDs = ds.filter(event -> "Jerry".equals(event.getUser()));

        ZhangsDs.print("zhangsan");
        LisiDs.print("Lisi");
        TomDs.print("Tom");
        JerryDs.print("Jerry");

         */

        // 使用侧输出流实现：
        // 创建输出标签

        OutputTag<Event> zhangsanOutputTag = new OutputTag<Event>("zhangsan", Types.POJO(Event.class));
        OutputTag<Event> lisiOutputTag = new OutputTag<Event>("lisi"){};
        OutputTag<Event> TomOutputTag = new OutputTag<Event>("Tom"){};

        SingleOutputStreamOperator<String> mainDs = ds.process(
                new ProcessFunction<Event, String>() {
                    @Override
                    public void processElement(Event value, ProcessFunction<Event, String>.Context context, Collector<String> out) throws Exception {

                        if ("Zhangs".equals((value.getUser()))) {
                            // 放入Zhangs流
                            context.output(zhangsanOutputTag, value);
                        } else if ("Lisi".equals(value.getUser())) {
                            // 放入Lisi流
                            context.output(lisiOutputTag, value);
                        } else if ("Tom".equals(value.getUser())) {
                            //放入Tom流
                            context.output(TomOutputTag, value);
                        } else {
                            //主流 Jerry  Peiqi
                            out.collect(value.toString());
                        }

                    }
                }
        );
        mainDs.print("main");

        // 通过主流获取侧流
        mainDs.getSideOutput(zhangsanOutputTag).print("zhangsan");
        mainDs.getSideOutput(lisiOutputTag).print("lisi");
        mainDs.getSideOutput(TomOutputTag).print("Tom");



        env.execute();


    }
}
